package org.dragonnova.iot.business.cloud.mq.service;

import java.util.List;
import java.util.Map;
import java.util.Random;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;

public class HashPartitioner implements Partitioner {

	private Map<String, ?> configs;

	@Override
	public void configure(Map<String, ?> configs) {
		this.configs = configs;
	}

	@Override
	public void close() {
		// TODO Auto-generated method stub

	}

	@Override
	public int partition(String topic, Object key, byte[] keyBytes,
			Object value, byte[] valueBytes, Cluster cluster) {
		List<PartitionInfo> numPar = cluster.partitionsForTopic(topic);
		// Integer k;
		// try {
		// k = Integer.parseInt((String) key);
		// } catch (NumberFormatException e) {
		// k = key.hashCode();
		// }
		// return Math.abs(k % numPar);
		int numPartitions = numPar.size();

		int randomNum = new Random().nextInt(numPartitions);

		return numPar.get(randomNum).partition();
	}

}
